package com.young.job;

import com.google.common.base.Stopwatch;
import com.google.common.util.concurrent.*;
import com.google.gson.GsonBuilder;
import com.sun.net.httpserver.HttpHandler;
import com.young.common.Config;
import com.young.common.Pair;
import com.young.common.YScheduler;
import com.young.context.ContextEnvironmentTool;
import com.young.context.InetTool;
import com.young.domain.CommunicationTypeEnum;
import com.young.domain.JobExecuteResultEnum;
import com.young.domain.JobScheduleStatusEnum;
import com.young.rpc.http.CallBackRequestBody;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.AnnotationUtils;
import com.young.rpc.handler.DefaultHttpJobCheckHandler;
import com.young.rpc.handler.DefaultHttpJobHandler;
import com.young.rpc.handler.JobHandler;
import com.young.rpc.http.HeartBeatRequestBody;
import com.young.rpc.http.ScheduleHttpClient;
import com.young.rpc.server.DefaultRpcServerImpl;
import com.young.rpc.server.RpcServer;

import javax.annotation.PostConstruct;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static com.young.common.Constants.*;

/**
 * @description：任务节点
 * @author: yangyahui01
 * @date: 11/25/24 2:49 PM
 */
public class JobNode implements ApplicationContextAware, JobHandler<JobExecuteResultEnum> {

    private static final Logger LOGGER = LoggerFactory.getLogger(JobNode.class);

    private ScheduleHttpClient scheduleHttpClient;


    public static final int DEFAULT_JOB_SERVICE_PORT = 50091; //client sdk默认的端口号

    private volatile int doJobServicePort = DEFAULT_JOB_SERVICE_PORT;


    private volatile List<ScheduleServerInstanceDO> serverNodes = new ArrayList<>();

    /**
     * scheduler-server 节点信息，是从starter赋值的
     */
    private volatile List<ScheduleServerInstanceDO> serverNodesDefault = new ArrayList<>();

    /**
     * 这里保存的是业务端的YScheduler任务信息
     */
    private ConcurrentHashMap<String, Object> localJobList = new ConcurrentHashMap<>();

    private static final GsonBuilder G_BUILDER = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    private final ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1);

    private ExecutorService executor = new ThreadPoolExecutor(Config.scheduleThreadInitCount,
            Config.scheduleThreadMaxCount, 30, TimeUnit.SECONDS, new SynchronousQueue<>());

    private ListeningExecutorService executorService = MoreExecutors.listeningDecorator(executor);

    private AtomicInteger serverIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Pair> jobFutures = new ConcurrentHashMap<>();



    private Map<String, Object> initedJobs;


    private String profile;

    private String cloud;

    private String group;

    private String appName;

    private RpcServer rpcServer;




    private ApplicationContext applicationContext;

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {

        this.applicationContext = applicationContext;
    }


    @PostConstruct
    public void init() throws Exception {

        serverNodes = serverNodesDefault;

        /**
         * 业务端服务的 profile、cloud、group
         */
        profile = ContextEnvironmentTool.preferProfile(applicationContext.getEnvironment());

        cloud = ContextEnvironmentTool.cloud(applicationContext.getEnvironment());

        group = ContextEnvironmentTool.preferGroup(applicationContext.getEnvironment());

        group = StringUtils.isBlank(group) ? profile : group;

        /**
         * 注册任务Job
         */
        if (initedJobs != null) {
            localJobList.putAll(initedJobs);
        }

        /**
         * 寻找标记了YScheduler的业务系统的bean
         */
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(YScheduler.class);

        for (Object obj : beans.values()) {

            Method[] methods = obj.getClass().getMethods();

            String className = obj.getClass().getSimpleName();

            if (AopUtils.isAopProxy(obj)) {
                className = obj.getClass().getSuperclass().getSimpleName();
            }

            for (Method method : methods) {
                /**
                 * 找到标记了YScheduler的方法
                 */
                YScheduler ySchedulerOnMethod = AnnotationUtils.findAnnotation(method, YScheduler.class);

                if (ySchedulerOnMethod != null) {
                    String methodName = method.getName();

                    String jobUniqCode = appName + "_" + className + "_" + methodName;

                    localJobList.put(jobUniqCode, obj);
                }
            }
        }

        LOGGER.info("init job finish ......{}", localJobList);


        Map<String, HttpHandler> serverHandlerMapping = new HashMap<>();

        serverHandlerMapping.put(CLIENT_DO_JOB_SERVER_PATH, new DefaultHttpJobHandler(this));

        serverHandlerMapping.put(CLIENT_JOB_CHECK_SERVER_PATH, new DefaultHttpJobCheckHandler(this));


        RpcServer.Builder builder = DefaultRpcServerImpl.builder();

        rpcServer = builder.port(doJobServicePort)
                .daemon(true)
                .handler(serverHandlerMapping).build();

        /**
         * 启动一个 HttpServer,监听 handler 请求
          */
        rpcServer.start();

        //开启心跳 1次/min
        scheduledExecutorService.scheduleWithFixedDelay(new HeartBeatHandler(), 10, 60, TimeUnit.SECONDS);
    }

    class HeartBeatHandler implements Runnable {
        private boolean registered = false;
        @Override
        public void run() {
            LOGGER.info("HeartBeatHandler fire");
            try {
                if(!registered) {
                    registered = register();
                    LOGGER.info("[young-schedule-client]节点注册结果->{}",registered ? "成功" : "失败");
                } else {
                    heartbeat();
                }
            } catch (Exception e) {
                LOGGER.error("应用[{}]连接[young-schedule-server]服务端失败", appName, e);
            }
        }
    }

    /**
     * 心跳请求
     */
    private void heartbeat() {
        ScheduleServerInstanceDO server = choose(serverNodes);
        String url = url(server, CLIENT_REGISTRY_URL_FORMATTER);
        LOGGER.info("向调度平台-{}发起young-schedule-client节点心跳","http://" + server.getIp() + ":" + server.getPort());
        sync(url,CommunicationTypeEnum.HEARTBEAT);
    }

    /**
     * 注册请求
     * @return
     * @throws Exception
     */
    private boolean register() throws Exception {
        ScheduleServerInstanceDO server = choose(serverNodesDefault);
        String url = url(server, CLIENT_REGISTRY_URL_FORMATTER);
        LOGGER.info("向调度平台-{}发起young-schedule-client节点注册","http://" + server.getIp() + ":" + server.getPort());
        return sync(url, CommunicationTypeEnum.REGISTER).get(6,TimeUnit.SECONDS).booleanValue();
    }

    /**
     * 发送请求
     * @param url
     * @param communicationTypeEnum
     * @return
     */
    private  Future<Boolean> sync(String url, CommunicationTypeEnum communicationTypeEnum) {
        HeartBeatRequestBody requestBody = new HeartBeatRequestBody();
        requestBody.setUrl(url);
        requestBody.setProject(appName);
        requestBody.setProfile(profile);
        requestBody.setCloud(cloud);
        requestBody.setGroup(group);
        requestBody.setHost(InetTool.HOST_NAME);
        requestBody.setIp(InetTool.LOCAL_IP);
        requestBody.setPort(doJobServicePort);
        requestBody.setCommunicationType(communicationTypeEnum.getType());
        StringBuilder stringBuilder = new StringBuilder();
        localJobList.keySet().forEach(job-> {
            if(stringBuilder.length() > 0) {
                stringBuilder.append(",");
            }
            stringBuilder.append(job);
        });
        requestBody.setJobs(stringBuilder.toString());
        return scheduleHttpClient.asyncSend(requestBody);
    }


    /**
     * 执行业务系统的调度方法
     * @param jobContext 任务上下文
     * @return
     */
    public JobExecuteResultEnum doJob(JobContext jobContext) {

        if (Objects.isNull(jobContext)) {
            LOGGER.error("young-schedule-client执行任务调度，jobContext参数为null");
            return JobExecuteResultEnum.CLIENT_ERROR;
        }

        if(jobContext.getCallBackNodes() == null || jobContext.getCallBackNodes().size() < 1) {
            LOGGER.error("young-schedule-client 执行任务调度，young-schedule-server回调节点为空");
            return JobExecuteResultEnum.CLIENT_ERROR;
        }

        //得到schedule-server节点列表
        serverNodes = jobContext.getCallBackNodes().stream()
                .filter(node -> StringUtils.isNoneBlank(node.getIp(), String.valueOf(node.getPort()))).collect(Collectors.toList());

        if(serverNodes.size() < 1) {
            serverNodes.addAll(serverNodesDefault);
            LOGGER.warn("young-schedule-client 执行任务调度，缺少有效的回调节点-{}",G_BUILDER.create().toJson(serverNodes,serverNodes.getClass()));
        }

        LOGGER.info("client doJob-jobUnicode:[{}] begin traceId=[{}] ", jobContext.getJobUniqCode(), jobContext.getTraceId());


        try {
            /**
             * jobUniqCode样例：starter-demo_TestTask_test
             */
            String jobMethod = jobContext.getJobUniqCode().split("_")[2];

            //标记了@YScheduler的Bean
            Object taskHandler = localJobList.get(jobContext.getJobUniqCode());

            if (null == taskHandler) {
                LOGGER.info("client doJob class not exist traceId={} ", jobContext.getTraceId());
                return JobExecuteResultEnum.CLIENT_NOT_EXIST;
            }


            //找到参数类型为JobContext的方法
            final Method method = taskHandler.getClass().getMethod(jobMethod, JobContext.class);

            if (Objects.isNull(method)) {
                LOGGER.info("client doJob method not exist traceId={} ", jobContext.getTraceId());
                return JobExecuteResultEnum.CLIENT_NOT_EXIST;
            }

            ListenableFuture<Long> future = executorService.submit(() -> {

                Stopwatch stopwatch = Stopwatch.createStarted();

                LOGGER.info("doJob future traceId={} ", jobContext.getTraceId());

                //去执行业务系统方法
                method.invoke(taskHandler, jobContext);

                long time = stopwatch.stop().elapsed(TimeUnit.SECONDS);
                LOGGER.info("任务执行时间：{} s", time);
                return time;
            });


            Pair pair = new Pair();
            pair.setFuture(future);
            pair.setJobContext(jobContext);
            jobFutures.put(jobContext.getTraceId(), pair);

            /**
             * 向 schedule-trigger 端发起回调请求
             */
            Futures.addCallback(future, new FutureCallback<Long>() {
                @Override
                public void onSuccess(Long duration) {
                    LOGGER.info("client callback future suc traceId={}", jobContext.getTraceId());

                    callBack(serverNodes, jobContext.getJobUniqCode(), jobContext.getTraceId(),
                            jobContext.getJobNodeNum(), jobContext.getJobNodeIndex(), JobScheduleStatusEnum.SUCCESS.getStatus(), duration, "");
                    jobFutures.remove(jobContext.getTraceId());

                }

                @Override
                public void onFailure(Throwable t) {

                    LOGGER.info("client callback future fail traceId=" + jobContext.getTraceId(), t);
                    String errMessage = ExceptionUtils.getStackTrace(t.getCause());
                    callBack(serverNodes, jobContext.getJobUniqCode(), jobContext.getTraceId(), jobContext.getJobNodeNum(), jobContext.getJobNodeIndex(), JobScheduleStatusEnum.FAIL.getStatus(), -1L, errMessage);
                    jobFutures.remove(jobContext.getTraceId());

                }
            }, MoreExecutors.directExecutor());


            if (future.isDone()) {
                jobFutures.remove(jobContext.getTraceId());
            }
        } catch (Exception e) {
            LOGGER.error("任务：[{}]执行出现异常", jobContext.getJobUniqCode(), e);
            return JobExecuteResultEnum.CLIENT_ERROR;
        }
        return JobExecuteResultEnum.SUCCESS;
    }

    private void callBack(List<ScheduleServerInstanceDO> serverInstances, String jobUniqCode, String traceId, Integer jobNodeNum,
                          Integer jobNodeIndex, Integer status, Long duration, String errorMessage) {
        int retryTimes = serverInstances.size();
        LOGGER.info("client callback tr，aceId={} rpc begin", traceId);
        do {
            ScheduleServerInstanceDO scheduleServer = choose(serverInstances);
            CallBackRequestBody callBackRequestBody = CallBackRequestBody.builder()
                    .url(url(scheduleServer, JOB_CALLBACK_URL_FORMATTER))
                    .jobUniqCode(jobUniqCode)
                    .traceId(traceId)
                    .callBackNode(scheduleServer.getIp())
                    .jobNodeNum(jobNodeNum)
                    .jobNodeIndex(jobNodeIndex)
                    .status(status)
                    .duration(duration)
                    .clientVersion(CLIENT_VERSION)
                    .errorMessage(errorMessage).build();

            if(scheduleHttpClient.send(callBackRequestBody)) {
                return;
            }
            retryTimes--;
        } while (retryTimes > 0);

        if(!serverInstances.equals(serverNodesDefault)) {
            LOGGER.info("client callback traceId={},重试{}次后启用默认调度节点={}发送回调请求",traceId,retryTimes,serverNodesDefault);
            callBack(serverNodesDefault,jobUniqCode,traceId,jobNodeNum,jobNodeIndex,status,duration,errorMessage);
        }

    }




    /**
     * scheduler-server端的节点选择策略
     * @param serverInstances
     * @return
     */
    private ScheduleServerInstanceDO choose(List<ScheduleServerInstanceDO> serverInstances) {
        int index = serverIndex.getAndIncrement() % serverInstances.size();
        if (serverIndex.get() >= serverInstances.size()) {
            serverIndex.set(0);
        }
        return serverInstances.get(index);
    }

    private String url(ScheduleServerInstanceDO scheduleServerInstanceDO, String urlFormatter) {
        return String.format(urlFormatter, scheduleServerInstanceDO.getIp(), scheduleServerInstanceDO.getPort());
    }


    public int check(JobContext jobContext) {
        String traceId = jobContext.getTraceId();
        LOGGER.info("checkRunningJob traceId={}", traceId);

        Pair fu = jobFutures.get(jobContext.getTraceId());

        if (null != fu) {
            return 0;
        }
        return -1;
    }


    public void setAppName(String appName) {
        this.appName = appName;
    }

    public void setDoJobServicePort(int doJobServicePort) {
        this.doJobServicePort = doJobServicePort;
    }

    public void setScheduleHttpClient(ScheduleHttpClient scheduleHttpClient) {
        this.scheduleHttpClient = scheduleHttpClient;
    }

    public void setServerNodesDefault(List<ScheduleServerInstanceDO> serverNodesDefault) {
        this.serverNodesDefault = serverNodesDefault;
    }
}
